package drds.global_transaction.transaction_coordinator.store;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import drds.common.$;
import drds.common.Constants;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.transaction_coordinator.transaction_session.TransactionSessionManager;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class FileTransactionStoreManager implements TransactionStoreManager
{
	public static final long max_enqueue_mills = 5 * 1000;
	public static final int max_thread_write = 1;
	public static final int max_shutdown_retry = 3;
	public static final int shutdown_check_internal = 1 * 1000;
	public static final int max_write_retry = 5;
	public static final String history_filename_postfix = ".1";
	public static final AtomicLong transaction_sequence_number = new AtomicLong(0);
	public static final AtomicLong file_flush_transaction_sequence_number = new AtomicLong(0);
	public static final int mark_size = 4;
	public static final int max_pool_time_mills = 2 * 1000;
	public static final int max_flush_time_mills = 2 * 1000;
	public static final int max_flush_num = 10;
	public static final boolean enable_schedule_flush = true;
	public static int per_file_block_size = 65535 * 8;
	public static long max_transaction_timeout_mills = 30 * 60 * 1000;
	public static volatile long startTimeMills = System.currentTimeMillis();
	public static long recoverCurrentOffset = 0;
	public static long recoverHisOffset = 0;
	//
	public BlockingQueue<TransactionWriteFuture> transactionWriteFutureQueue = new LinkedBlockingQueue<TransactionWriteFuture>();
	public volatile boolean stop = false;
	public File currentFile;
	public RandomAccessFile currentRandomAccessFile;
	public FileChannel currentFileChannel;
	public TransactionSessionManager transactionSessionManager;
	public String currentFileFullName;
	//
	public String historyFullFileName;
	private ExecutorService executorService;

	public FileTransactionStoreManager(TransactionSessionManager transactionSessionManager, String fullFileName) throws IOException
	{
		initFile(fullFileName);
		executorService = new ThreadPoolExecutor(//
				max_thread_write, //
				max_thread_write, //
				Integer.MAX_VALUE, //
				TimeUnit.MILLISECONDS, //
				new LinkedBlockingQueue<Runnable>(), //
				new NamedThreadFactory("fileTransactionStore", max_thread_write, true));
		executorService.submit(new WriteDataFileRunnable(this));
		this.transactionSessionManager = transactionSessionManager;
	}

	public void initFile(String currentFileFullName) throws IOException
	{
		this.currentFileFullName = currentFileFullName;
		this.historyFullFileName = currentFileFullName + history_filename_postfix;
		try
		{
			currentFile = new File(this.currentFileFullName);
			if (!currentFile.exists())
			{
				currentFile.createNewFile();
				startTimeMills = System.currentTimeMillis();
			} else
			{
				startTimeMills = currentFile.lastModified();
			}
			currentRandomAccessFile = new RandomAccessFile(currentFile, "rw");
			currentRandomAccessFile.seek(currentFile.length());
			currentFileChannel = currentRandomAccessFile.getChannel();
		} catch (IOException exx)
		{
			log.error("init file error," + exx.getMessage());
			throw exx;
		}
	}

	@Override
	public boolean saveOperation(SessionStorable sessionStorable, OperationType operationType)
	{
		TransactionWriteFuture transactionWriteFuture = new TransactionWriteFuture(sessionStorable, operationType);
		try
		{
			if (transactionWriteFutureQueue.offer(transactionWriteFuture, max_enqueue_mills, TimeUnit.MILLISECONDS))
			{
				return transactionWriteFuture.get();
			}
		} catch (InterruptedException exx)
		{
			log.error("write data file error," + exx.getMessage());
		}
		return false;
	}

	@Override
	public void shutdown()
	{
		if (null != executorService)
		{
			stop = true;
			executorService.shutdown();
			int retry = 0;
			while (!executorService.isTerminated() && retry < max_shutdown_retry)
			{
				++retry;
				try
				{
					Thread.sleep(shutdown_check_internal);
				} catch (InterruptedException exx)
				{
				}
			}
			if (retry >= max_shutdown_retry)
			{
				executorService.shutdownNow();
			}
		}
	}

	@Override
	public List<TransactionWriteStore> getTransactionWriteStoreListFromFile(int readSize, boolean isHistory)
	{
		File file = null;
		long currentOffset = 0;
		if (isHistory)
		{
			file = new File(historyFullFileName);
			currentOffset = recoverHisOffset;
		} else
		{
			file = new File(currentFileFullName);
			currentOffset = recoverCurrentOffset;
		}
		if (file.exists())
		{
			return parseDataFile(file, readSize, currentOffset);
		}
		return null;
	}

	@Override
	public boolean hasRemaining(boolean isHistory)
	{
		File file = null;
		RandomAccessFile randomAccessFile = null;
		long currentOffset = 0;
		if (isHistory)
		{
			file = new File(historyFullFileName);
			currentOffset = recoverHisOffset;
		} else
		{
			file = new File(currentFileFullName);
			currentOffset = recoverCurrentOffset;
		}
		try
		{
			randomAccessFile = new RandomAccessFile(file, "r");
			return currentOffset < randomAccessFile.length();
		} catch (IOException exx)
		{
		} finally
		{
			closeFile(null, randomAccessFile);
		}
		return false;
	}

	private List<TransactionWriteStore> parseDataFile(File file, int readSize, long currentOffset)
	{
		List<TransactionWriteStore> transactionWriteStoreList = new ArrayList<>(readSize);
		RandomAccessFile randomAccessFile = null;
		FileChannel fileChannel = null;
		try
		{
			randomAccessFile = new RandomAccessFile(file, "r");
			randomAccessFile.seek(currentOffset);
			fileChannel = randomAccessFile.getChannel();
			fileChannel.position(currentOffset);
			long size = randomAccessFile.length();
			ByteBuffer byteBuffer = ByteBuffer.allocate(mark_size);
			while (fileChannel.position() < size)
			{
				try
				{
					byteBuffer.clear();
					int read = fileChannel.read(byteBuffer);
					if (read != mark_size)
					{
						break;
					}
					byteBuffer.flip();
					int bodySize = byteBuffer.getInt();
					byte[] bodyBytes = new byte[bodySize];// a+b模式
					ByteBuffer bodyByteBuffer = ByteBuffer.wrap(bodyBytes);
					read = fileChannel.read(bodyByteBuffer);
					if (read != bodySize)
					{
						break;
					}
					TransactionWriteStore transactionWriteStore = new TransactionWriteStore();
					transactionWriteStore.decode(bodyBytes);
					transactionWriteStoreList.add(transactionWriteStore);
					if (transactionWriteStoreList.size() == readSize)
					{
						break;
					}
				} catch (Exception ex)
				{
					log.error("decode data file error:" + ex.getMessage());
					break;
				}
			}
			return transactionWriteStoreList;
		} catch (IOException exx)
		{
			log.error("parse data file error:" + exx.getMessage() + ",file:" + file.getName());
			return null;
		} finally
		{
			try
			{
				if (null != fileChannel)
				{
					if (isHistoryFile(file))
					{
						recoverHisOffset = fileChannel.position();
					} else
					{
						recoverCurrentOffset = fileChannel.position();
					}
				}
				closeFile(fileChannel, randomAccessFile);
			} catch (IOException exx)
			{
				log.error("file close error," + exx.getMessage());
			}
		}

	}

	private boolean isHistoryFile(File file)
	{
		return file.getName().endsWith(history_filename_postfix);
	}

	public void closeFile(FileChannel fileChannel, RandomAccessFile randomAccessFile)
	{
		try
		{
			if (null != fileChannel)
			{
				fileChannel.close();
				fileChannel = null;
			}
			if (null != randomAccessFile)
			{
				randomAccessFile.close();
				randomAccessFile = null;
			}
		} catch (IOException exx)
		{
			if (Constants.developMode)
			{
				exx.printStackTrace();
			}
			log.error($.printStackTraceToString(exx));
			log.error("file close error," + exx.getMessage());
		}
	}

}
